package com.gugee.ins.data.task.blogger.service;

import com.gugee.common.task.dto.TaskParam;
import com.gugee.common.task.model.TaskRecord;
import com.gugee.common.task.service.AbstractProducerHandler;
import com.gugee.ins.common.constant.InsBloggerMsgTypeCodeEnum;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.blogger.InsBloggerMainDao;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.gugee.ins.data.task.config.TaskConfig;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: luolifeng
 * Date: 2020-12-24 18:32
 */
@Component
@Slf4j
public class InsBloggerUpdateSvc extends AbstractProducerHandler<InsBloggerMessage> {
    @Autowired
    @Qualifier("insBloggerMsgBaseInfoProducer")
    private KafkaTemplate<Long, InsBloggerMessage> insBloggerMsgBaseInfoProducer;
    @Autowired
    public InsBloggerDao insBloggerDao;

    @Autowired
    public TaskConfig taskConfig;

    @Override
    public List<InsBloggerMessage> findMessageObject(TaskRecord taskRecord, TaskParam taskParam) {

        long startId=(taskRecord.getCurrentPage()-1)*taskParam.getRecords()+1L;
        long endId=(long)taskRecord.getCurrentPage()*taskParam.getRecords();
        long lastCollectTime= System.currentTimeMillis() / 1000 - taskConfig.getIntervalHours() * 3600;
        List<InsBloggerMessage> rst= insBloggerDao.findInsBloggerMessageByPages(startId,endId,taskConfig.getInsBloggerMainFollowerCountMin(),lastCollectTime);
        return rst==null?new ArrayList<>():rst;
    }

    /**
     * 查询当前有最大Blogger的id值，用于代表记录数
     * @return
     */
    @Override
    public long objectTotals() {
        return insBloggerDao.getMaxId();
    }

    @Override
    public void sendMsg(List<InsBloggerMessage> messages, String topic) {
        messages.forEach(msg->{
            msg.setRetryTimes(taskConfig.getRetryBloggerTimes());
            msg.setRetryLimitTimes(taskConfig.getRetryBloggerTimes());
            msg.setTypeCode(InsBloggerMsgTypeCodeEnum.UID_UPDATE_F_BLOGGER_TIMING.getCode());
            insBloggerMsgBaseInfoProducer.send(topic,msg);
        });
        XxlJobLogger.log("投递主题={},投递消息数={}", topic,messages.size());
        log.info("投递主题={},投递消息数={}", topic,messages.size());
    }
}
